package cn.spark.study.sql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import java.util.List;

/**
 * 通过反射的方式将RDD转换为DataFrame
 *
 * @author jun.zhang6
 * @date 2020/11/13
 */
public class RDD2DataFrameReflection {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setMaster("local")
                .setAppName("RDD2DataFrameReflection");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);

        JavaRDD<String> lines = sc.textFile("C:\\Users\\jun.zhang6\\Desktop\\students.txt");

        JavaRDD<Student> students = lines.map(new Function<String, Student>() {
            @Override
            public Student call(String line) throws Exception {
                String[] lineSplited = line.split(",");
                Student student = new Student();
                student.setId(Integer.parseInt(lineSplited[0]));
                student.setName(lineSplited[1]);
                student.setAge(Integer.parseInt(lineSplited[2]));
                return student;
            }
        });

        //通过反射获取，Student类必须序列化
        DataFrame studentDF = sqlContext.createDataFrame(students, Student.class);

        //注册成一个临时表
        studentDF.registerTempTable("students");

        //然后针对表执行SQL
        DataFrame teenagerDF = sqlContext.sql("select * from students where age <= 18");

        //将DataFrame转为RDD
        JavaRDD<Row> teenagerRDD = teenagerDF.javaRDD();

        //将RDD的数据映射为Student
        JavaRDD<Student> teenagerStudentRDD = teenagerRDD.map(new Function<Row, Student>() {
            @Override
            public Student call(Row row) throws Exception {
                //row中的数据顺序可能和期望的不一样
                Student student = new Student();
                student.setAge(row.getInt(0));
                student.setId(row.getInt(1));
                student.setName(row.getString(2));
                return student;
            }
        });

        //将数据collect回来
        List<Student> list = teenagerStudentRDD.collect();

        for (Student stu : list) {
            System.out.println(stu);
        }

        sc.close();
    }
}
